package cn.doitedu;

import org.apache.hadoop.hive.ql.exec.UDAF;
import org.apache.hadoop.hive.ql.exec.UDAFEvaluator;
import org.roaringbitmap.longlong.Roaring64Bitmap;

import java.io.IOException;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/4/24
 * @Desc: 学大数据，上多易教育
 *   把一组整数聚合成一个bitmap
 **/
public class AggBigIntToBitmapUDAF extends UDAF {


    public static class AggBigIntToBitmapEvaluator implements UDAFEvaluator {

        Acc acc;

        public AggBigIntToBitmapEvaluator() {
            init();
        }


        // 初始化中间累加器
        @Override
        public void init() {
            try {
                acc = new Acc();
                System.out.println("累加器初始化了....................");
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }



        // 逐条聚合
        public boolean iterate(Long element) throws IOException {

            if(element != null ) {
                // 取出累加器中的bitmap
                Roaring64Bitmap bitmap = this.acc.getBitmap();
                // 向bitmap中添加元素
                bitmap.add(element);

                System.out.println("iterate  收到一条数据: " + element);
                System.out.println("iterate  此刻的bitmap是: " + bitmap);

                // 把添加元素后的bitmap重新设置到累加器中
                this.acc.setBitmap(bitmap);
            }
            return true;
        }


        // map端输出的局部结果
        public Acc terminatePartial() {
            return this.acc;
        }


        // reduce端合并map端输出的acc
        public boolean merge(Acc otherAcc) throws IOException {

            // 取出自己的累加器中的bitmap
            Roaring64Bitmap thisBitmap = this.acc.getBitmap();

            // 取出传入的另一个累加器中的bitmap
            Roaring64Bitmap otherBitmap = otherAcc.getBitmap();


            System.out.println("merge,  thisBitmap:   " + thisBitmap );
            System.out.println("merge,  otherBitmap:   " + otherBitmap );


            // 合并两个bitmap到thisBitmap
            thisBitmap.or(otherBitmap);


            System.out.println("merge,  合并完之后:   " + thisBitmap );


            // 把合并后的bitmap重新设置到累加器中
            this.acc.setBitmap(thisBitmap);

            return true;
        }

        // 输出最终结果
        public byte[] terminate() {
            return this.acc.getBitmapBytes();
        }
    }

}
